package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.StockBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


public class SecStockWindowFunction implements WindowFunction<CleanBean, StockBean,String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow timeWindow, Iterable<CleanBean> iterable, Collector<StockBean> collector) throws Exception {
        // 记录最新个股，每5s生成一条行情数据(事件时间最大）
        CleanBean cleanBean = null;
        for (CleanBean line : iterable) {
            if(cleanBean==null){
                cleanBean=line;
            }
            //比较事件时间
            if(line.getEventTime()>cleanBean.getEventTime()){
                cleanBean=line;
            }
        }
        //格式化日期
        Long timeTransfer = DateUtil.longTimeTransfer(cleanBean.getEventTime(), Constant.format_YYYYMMDDHHMMSS);//秒级行情

        //封装输出数据
        StockBean stockBean = new StockBean();
        stockBean.setEventTime(cleanBean.getEventTime());
        stockBean.setSecCode(cleanBean.getSecCode());
        stockBean.setSecName(cleanBean.getSecName());
        stockBean.setPreClosePrice(cleanBean.getPreClosePrice());
        stockBean.setOpenPrice(cleanBean.getOpenPrice());
        stockBean.setClosePrice(cleanBean.getTradePrice());
        stockBean.setHighPrice(cleanBean.getMaxPrice());
        stockBean.setLowPrice(cleanBean.getMinPrice());
        stockBean.setTradeVol(0L); //分时成交量
        stockBean.setTradeAmt(0L);
        stockBean.setTradeVolDay(cleanBean.getTradeVolumn());
        stockBean.setTradeAmtDay(cleanBean.getTradeAmt());
        stockBean.setTradeTime(timeTransfer);
        stockBean.setSource(cleanBean.getSource());

        collector.collect(stockBean);

    }

}
